/*
 * $Id$
 * --------------------------------------------------------------------------------------
 * Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com
 *
 * The software in this package is published under the terms of the CPAL v1.0
 * license, a copy of which has been included with this distribution in the
 * LICENSE.txt file.
 */

package org.mule.transport.jpa;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;
import javax.persistence.Query;

import org.apache.commons.lang.StringUtils;
import org.mule.DefaultMuleMessage;
import org.mule.api.MuleEvent;
import org.mule.api.MuleMessage;
import org.mule.api.endpoint.OutboundEndpoint;
import org.mule.api.transaction.Transaction;
import org.mule.api.transformer.TransformerException;
import org.mule.transaction.TransactionCoordination;
import org.mule.transport.AbstractMessageDispatcher;
import org.mule.transport.jpa.exceptions.BadConfigurationException;
import org.mule.transport.jpa.strategy.EntityManagerStrategy;
import org.mule.transport.jpa.strategy.EntityManagerStrategyFactory;
import org.mule.util.ArrayUtils;
import org.mule.util.MapUtils;

import com.mysema.query.jpa.impl.JPAQuery;

/**
 * <code>JpaMessageDispatcher</code> TODO document
 */
public class JpaMessageDispatcher extends AbstractMessageDispatcher {

	private static final String QUERY_DSL_ACTION = "queryDSL";

	private static final int DEFAULT_MAX_FETCH_SIZE = 100;

	protected JpaConnector jpaConnector;

	private List<?> paramNames;
	private String queryString;

	private Integer maxResults;

	/*
	 * For general guidelines on writing transports see
	 * http://www.mulesoft.org/documentation
	 * /display/MULE3USER/Creating+Transports
	 */

	public JpaMessageDispatcher(OutboundEndpoint endpoint) {
		super(endpoint);
		jpaConnector = (JpaConnector) endpoint.getConnector();
		maxResults = MapUtils.getInteger(endpoint.getProperties(), JpaConnector.PROPERTY_MAX_FETCH_SIZE, DEFAULT_MAX_FETCH_SIZE);
		parseStatement(jpaConnector.getStatement(endpoint));
	}

	protected void parseStatement(String statement) {
		this.paramNames = new ArrayList<Object>();
		this.queryString = this.jpaConnector.parseStatement(statement, this.paramNames);
	}

	@Override
	public void doConnect() throws Exception {
		/*
		 * IMPLEMENTATION NOTE: Makes a connection to the underlying resource.
		 * Where connections are managed by the connector this method may do
		 * nothing
		 */

		// If a resource for this Dispatcher needs a connection established,
		// then this is the place to do it
	}

	@Override
	public void doDisconnect() throws Exception {
		/*
		 * IMPLEMENTATION NOTE: Disconnect any conections made in the connect
		 * method
		 */

		// If the connect method did not do anything then this method
		// shouldn't do anything either
	}

	@Override
	public void doDispatch(MuleEvent event) throws Exception {
		/*
		 * IMPLEMENTATION NOTE: This is invoked when the endpoint is
		 * asynchronous. It should invoke the transport but not return any
		 * result. If a result is returned it should be ignorred, but if the
		 * underlying transport does have a notion of asynchronous processing,
		 * that should be invoked. This method is executed in a different thread
		 * to the request thread.
		 */

		/*
		 * IMPLEMENTATION NOTE: The event message needs to be transformed for
		 * the outbound transformers to take effect. This isn't done
		 * automatically in case the dispatcher needs to modify the message
		 * before apllying transformers. To get the transformed outbound message
		 * call - event.transformMessage();
		 */

		logger.debug("Dispatch event: " + event);
		doSend(event);
	}

	@Override
	public MuleMessage doSend(MuleEvent event) throws Exception {

		/*
		 * IMPLEMENTATION NOTE: Should send the event payload over the
		 * transport. If there is a response from the transport it shuold be
		 * returned from this method. The sendEvent method is called when the
		 * endpoint is running synchronously and any response returned will
		 * ultimately be passed back to the callee. This method is executed in
		 * the same thread as the request thread.
		 */

		/*
		 * IMPLEMENTATION NOTE: The event message needs to be transformed for
		 * the outbound transformers to take effect. This isn't done
		 * automatically in case the dispatcher needs to modify the message
		 * before apllying transformers. To get the transformed outbound message
		 * call - event.transformMessage();
		 */

		MuleMessage msg = null;
		logger.debug("Property " + queryString + " :" + event.getMessage().findPropertyInAnyScope(queryString, null));
		if (EntityManagerStrategyFactory.isValidReceiveAction(queryString)) {
			msg = doExecute(event, EntityManagerStrategyFactory.createEntityManagerCommand(jpaConnector.getEntityManager(), queryString));
		} else if (queryString.equals(QUERY_DSL_ACTION)) {
			if (!(event.getMessage().getPayload() instanceof QueryDslQuery)) {
				throw new BadConfigurationException("Payload is not a QueryDSL Query although the endpoint is configured to do that");
			}
			msg = doQueryWithQueryDsl(event, (QueryDslQuery) event.getMessage().getPayload());
		} else {
			msg = doQuery(event);
		}
		return msg;
	}

	// Query using querydsl language
	private MuleMessage doQueryWithQueryDsl(MuleEvent event, QueryDslQuery jpaQuery) throws Exception {
		EntityManager em = null;
		EntityTransaction etx = null;
		Transaction tx = TransactionCoordination.getInstance().getTransaction();
		try {
			em = jpaConnector.getEntityManager();
			if (tx == null) {
				etx = em.getTransaction();
			}
			if (!em.isOpen()) {
				throw new Exception("Cannot open entityManager.");
			}
			if (etx != null && !etx.isActive()) {
				etx.begin();
			}
			JPAQuery query = jpaQuery.getQuery().clone(em);
			List<?> result = query.limit(maxResults).list(jpaQuery.getBasePath());
			if (etx != null && etx.isActive()) {
				etx.commit();
			}
			return new DefaultMuleMessage(result, event.getMuleContext());
		} catch (Exception e) {
			logger.error("Error executing JPA query", e);
			if (etx != null && etx.isActive()) {
				etx.rollback();
			}
			throw e;
		}
	}

	/**
	 * merge to entitymanager.
	 * 
	 * @param event
	 * @return muleMessage
	 * @throws TransformerException
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	private MuleMessage doExecute(MuleEvent event, EntityManagerStrategy emStrategy) throws Exception {
		EntityManager em = null;
		EntityTransaction etx = null;
		Object payload = event.getMessage().getPayload();
		synchronized (this) {
			Transaction tx = TransactionCoordination.getInstance().getTransaction();
			try {
				em = emStrategy.getEntityManager();
				if (tx == null) {
					etx = em.getTransaction();
				}
				// em.persist(event.getMessage().getPayload());
				if (!em.isOpen()) {
					throw new Exception("Cannot open entityManager.");
				}
				if (etx != null && !etx.isActive()) {
					etx.begin();
				}
				Object result = null;
				if (payload instanceof List) {
					logger.debug("payload type is list.");
					List<Object> payloadList = (List<Object>) payload;
					List<Object> resultList = new ArrayList<Object>();
					for (Iterator<Object> payloadIter = payloadList.iterator(); payloadIter.hasNext();) {
						Object entity = payloadIter.next();
						logger.debug("entity is " + entity.getClass() + ".");
						Object iterResult = emStrategy.execute(entity);
						resultList.add(iterResult);
					}
					result = resultList;
				} else {
					logger.debug("entity is " + payload.getClass() + ".");
					result = emStrategy.execute(payload);
				}
				if (etx != null && etx.isActive()) {
					etx.commit();
				}
				event.getMessage().setPayload(result);
				return event.getMessage();
			} catch (Exception e) {
				logger.error("Error executing JPA query", e);
				if (etx != null && etx.isActive()) {
					etx.rollback();
				}
				throw e;
			}
		}
	}

	/**
	 * query to entityManager.
	 * 
	 * @param event
	 * @return muleMessage
	 * @throws TransformerException
	 */
	private MuleMessage doQuery(MuleEvent event) throws Exception {
		EntityManager em = null;
		EntityTransaction etx = null;
		Object payload = event.getMessage().getPayload();
		try {
			Transaction tx = TransactionCoordination.getInstance().getTransaction();
			em = jpaConnector.getEntityManager();
			if (tx == null) {
				etx = em.getTransaction();
			}
			if (!em.isOpen()) {
				throw new Exception("Cannot open entityManager.");
			}
			if (etx != null && !etx.isActive()) {
				etx.begin();
			}
			List<?> resultSet = null;
			// first Parse param names
			String dynamicallyParsedQuery = mapDynamicParams(event);
			Query query = em.createQuery(dynamicallyParsedQuery);
			// then search in payload
			if (payload instanceof Map) {
				Map<?, ?> payloadMap = (Map<?, ?>) payload;
				for (Object key : payloadMap.keySet()) {
					Object value = payloadMap.get(key);
					logger.debug("Param " + key + ": " + value + "(" + value.getClass() + ")");
					query.setParameter((String) key, value);
				}
			}
			resultSet = query.getResultList();
			if (etx != null && etx.isActive()) {
				etx.commit();
			}
			return new DefaultMuleMessage(resultSet, event.getMuleContext());
		} catch (Exception e) {
			logger.error("Error querying with JPA Transport", e);
			if (etx != null && etx.isActive()) {
				etx.rollback();
			}
			throw e;
		}
	}

	private String mapDynamicParams(MuleEvent event) throws Exception {
		Object[] readParams = jpaConnector.getParams(endpoint, this.paramNames, event, this.endpoint.getEndpointURI().getAddress());
		String temporalQuery = queryString;
		if (logger.isDebugEnabled()) {
			logger.debug("SQL QUERY: " + temporalQuery + ", params = " + ArrayUtils.toString(readParams));
		}
		int i = 0;
		while (StringUtils.indexOf(temporalQuery, "?") != -1) {
			temporalQuery = StringUtils.replaceOnce(temporalQuery, "?", readParams[i].toString());
			i++;
		}
		return temporalQuery;
	}

	@Override
	public void doDispose() {
		// Optional; does not need to be implemented. Delete if not required
		/*
		 * IMPLEMENTATION NOTE: Is called when the Dispatcher is being disposed
		 * and should clean up any open resources.
		 */
	}
}
